/*
 * @(#)MultiThreadExecutor.java 1.0 2014/02/14
 *
 * Copyright (c) 2014 SinoSoft Refactored by filon51.
 * All rights reserved.
 *
 */
package com.sinosoft.cloud.dp.thread;

//import EDU.oswego.cs.dl.util.concurrent.BoundedLinkedQueue;
//import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
//import EDU.oswego.cs.dl.util.concurrent.Semaphore;

import org.apache.log4j.Logger;

import java.util.HashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.Semaphore;


/**
 * 多线程执行引擎
 *
 * @author filon51
 * @since 2014-02-14
 */
public class MultiThreadExecutor implements ThreadExecutor {
	private Logger m_log = Logger.getLogger(MultiThreadExecutor.class);
	/** 设置数据处理的线程数据，真正的线程数还要加上数据读取线程和数据写入线程。 */
	private int m_processThreadCount = 8;
//	private BoundedLinkedQueue m_readDataQueue = new BoundedLinkedQueue(1024);
//	private BoundedLinkedQueue m_writeDataQueue = new BoundedLinkedQueue(1024);
	private LinkedBlockingDeque m_readDataQueue = new LinkedBlockingDeque(1024);
	private LinkedBlockingDeque m_writeDataQueue = new LinkedBlockingDeque(1024);
	private Semaphore m_semaphoreFinish = new Semaphore(0);

	private static HashMap OBJ_READ_GUARD = new HashMap();
	private static HashMap OBJ_WRITE_GUARD = new HashMap();
	private static boolean EXCEPTION_FLAG = false;
	private static HashMap OBJ_EXCEPTION_GUARD = new HashMap();

	public MultiThreadExecutor() {
	}

	public void executeService(BusinessService businessService) {
		if (businessService == null) {
			m_log.fatal("parameter BusinessService can't be null");
			return;
		}

		m_log.debug("BusinessService start");
		businessService.onInitialiseResource();

		/*PooledExecutor executor = new PooledExecutor();
		executor.setMaximumPoolSize(getM_processThreadCount() + 2);
		executor.setMinimumPoolSize(getM_processThreadCount() + 2);
		executor.setKeepAliveTime(1000);*/
		ExecutorService executor = Executors.newFixedThreadPool(getM_processThreadCount()+2);

		try {
			executor.execute(new ReaderThread(businessService,
					m_readDataQueue, getM_processThreadCount()));

			for (int nIndex = 0; nIndex < getM_processThreadCount(); nIndex++) {
				Thread thread = new ProcessorThread(businessService,
						m_readDataQueue, m_writeDataQueue);
				thread.setName("thread_" + String.valueOf(nIndex));
				executor.execute(thread);
			}

			executor.execute(new WriterThread(businessService,
					m_writeDataQueue, m_semaphoreFinish, getM_processThreadCount()));
			// wait for all dp to finish
			m_semaphoreFinish.acquire();
		} catch (Exception ex) {
			m_log.error("main loop", ex);
			ex.printStackTrace();
		} finally {
			executor.shutdown();
//			executor.shutdownAfterProcessingCurrentlyQueuedTasks();
			businessService.onReleaseResource();// 释放资源
		}
		m_log.debug("BusinessService end");
	}

	public static void setOBJ_READ_GUARD(HashMap oBJ_READ_GUARD) {
		OBJ_READ_GUARD = oBJ_READ_GUARD;
	}

	public static HashMap getOBJ_READ_GUARD() {
		return OBJ_READ_GUARD;
	}

	public static void setOBJ_WRITE_GUARD(HashMap oBJ_WRITE_GUARD) {
		OBJ_WRITE_GUARD = oBJ_WRITE_GUARD;
	}

	public static HashMap getOBJ_WRITE_GUARD() {
		return OBJ_WRITE_GUARD;
	}

	public static HashMap getOBJ_EXCEPTION_GUARD() {
		return OBJ_EXCEPTION_GUARD;
	}

	public static void setOBJ_EXCEPTION_GUARD(HashMap oBJ_EXCEPTION_GUARD) {
		OBJ_EXCEPTION_GUARD = oBJ_EXCEPTION_GUARD;
	}

	/**
	 * 异常标记 默认FALSE；如有某一个线程出现异常，将改写此全局成员变量为TRUE
	 */
	public static boolean isExceptionFlag() {
		return EXCEPTION_FLAG;
	}

	public static void setExceptionFlag(boolean exceptionFlag) {
		EXCEPTION_FLAG = exceptionFlag;
	}

	/** 设置数据处理的线程数据，真正的线程数还要加上数据读取线程和数据写入线程。 */
	public int getM_processThreadCount() {
		return m_processThreadCount;
	}

	public void setM_processThreadCount(int m_processThreadCount) {
		this.m_processThreadCount = m_processThreadCount;
	}
}